package com.deep.flink.demo.job._3_function;

import lombok.SneakyThrows;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import study.bigdata.flink.cdc.function.AggFunction;

public class _3_AggregateFunctionJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        //========== 【注册函数】 ==========
        tableEnv.createTemporarySystemFunction("test", AggFunction.class);
        //
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE order_base (")
                .append("rb_id      BIGINT,")
                .append("rb_ub_id   BIGINT,")
                .append("rb_amount  DOUBLE,")
                .append("PRIMARY KEY (rb_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'mysql-cdc', ")
                .append("'hostname' = '127.0.0.1', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'database-name' = 'ds_canal', ")
                .append("'table-name'= 'order_base', ")
                .append("'port' = '3306', ")
                .append("'server-id' = '1', ")
                .append("'server-time-zone' = 'Asia/Shanghai', ")
                .append("'debezium.database.whitelist' = 'ds_canal', ")
                .append("'debezium.table.whitelist' = 'ds_canal.order_base', ")
                .append("'debezium.include.schema.changes' = 'false', ")
                .append("'debezium.snapshot.locking.mode' = 'none'")
                .append(")");
        TableResult tblResult = tableEnv.executeSql(ddl.toString());
        //========== 【查询表】 ==========
        StringBuffer sql = new StringBuffer();
        sql.append("select rb_ub_id, ")
                .append("test(rb_amount) ")
                .append("from order_base ")
                .append("group by rb_ub_id ");
        Table table = tableEnv.sqlQuery(sql.toString());
        //
        tableEnv.toRetractStream(table, Row.class).print(">>>>>>>>>>>>>>>>>>>>");

        //
        env.execute("cdc-debezium-query-job");
    }
}
